package kafka2

import (
	"gitee.com/chejiangyi/bsfgo/core/base2"
	"github.com/IBM/sarama"
	"time"
)

// KafkaConfig 定义Kafka客户端配置参数
type KafkaConfig struct {
	// Brokers Kafka broker地址列表，格式为 ["host1:port1", "host2:port2"]
	// 必填项，无默认值
	Brokers []string

	// Version Kafka集群版本号，必须与服务器版本兼容
	// 示例: "2.5.0"、"2.8.1"等
	// 建议默认值: "2.8.0" (当前广泛使用的稳定版本)
	Version string

	// GroupID 消费者组ID，用于消费者分组和偏移量管理
	// 消费者必填项，生产者可留空
	GroupID string

	// ============ 生产者配置 ============

	// ProducerRequiredAck 生产者要求broker返回的确认级别
	// 建议默认值: sarama.WaitForLocal (兼顾可靠性和性能)
	ProducerRequiredAck sarama.RequiredAcks

	// ProducerMaxMessageBytes 单个消息最大字节数(包括key和value)
	// 建议默认值: 1000000 (1MB)，不能超过broker的message.max.bytes配置
	ProducerMaxMessageBytes int

	// ProducerCompression 消息压缩算法
	// 建议默认值: sarama.CompressionSnappy (良好的压缩比和性能平衡)
	ProducerCompression sarama.CompressionCodec

	// ProducerReturnSuccesses 是否返回成功发送的消息信息
	// 建议默认值: false (除非需要严格确认每条消息发送状态)
	ProducerReturnSuccesses bool

	// ============ 消费者配置 ============

	// ConsumerInitialOffset 消费者初始偏移量设置
	// 建议默认值: sarama.OffsetNewest (从最新消息开始消费)
	ConsumerInitialOffset int64

	// ConsumerAssignor 消费者分区分配策略
	// 建议默认值: "range" (与旧版本兼容) 或 "sticky" (减少再平衡时的分区移动)
	ConsumerAssignor string

	// ConsumerReturnErrors 是否返回消费过程中的错误
	// 建议默认值: true (推荐监控消费错误)
	ConsumerReturnErrors bool

	// ConsumerOffsetsAutoCommitEnable 是否自动提交偏移量
	// 建议默认值:
	//   - 高可靠性要求: false (手动提交)
	//   - 普通场景: true (自动提交)
	ConsumerOffsetsAutoCommitEnable bool

	// ConsumerOffsetsAutoCommitInterval 自动提交偏移量的时间间隔
	// 建议默认值: 5*time.Second (需要根据消息处理速度调整)
	ConsumerOffsetsAutoCommitInterval time.Duration

	// ConsumerMaxProcessingTime 处理单条消息的最大允许时间
	// 建议默认值: 30*time.Second (根据业务处理时间调整)
	ConsumerMaxProcessingTime time.Duration
}

func (config KafkaConfig) CheckConfig(cfg *sarama.Config) {
	// 版本配置
	if config.Version != "" {
		version, err := sarama.ParseKafkaVersion(config.Version)
		if err != nil {
			panic(base2.NewBsfError2("无效 kafka version: %w", err))
		}
		cfg.Version = version
	}
}

func (config KafkaConfig) parseRequiredAcks(val string) sarama.RequiredAcks {
	switch val {
	case "WaitForAll":
		return sarama.WaitForAll
	case "WaitForLocal":
		return sarama.WaitForLocal
	case "NoResponse":
		return sarama.NoResponse
	default:
		return sarama.WaitForLocal
	}
}

func (config KafkaConfig) parseProducerCompression(val string) sarama.CompressionCodec {
	switch val {
	case "none":
		return sarama.CompressionNone
	case "gzip":
		return sarama.CompressionGZIP
	case "snappy":
		return sarama.CompressionSnappy
	case "lz4":
		return sarama.CompressionLZ4
	case "zstd":
		return sarama.CompressionZSTD
	default:
		return sarama.CompressionNone
	}
}

func (*KafkaUtilTool) parseConsumerInitialOffset(val string) int64 {
	switch val {
	case "OffsetNewest":
		return sarama.OffsetNewest
	case "OffsetOldest":
		return sarama.OffsetOldest
	default:
		return sarama.OffsetNewest
	}
}
